27. 中国零售客户 RFM 探索（跨国样例对照）

import pandas as pd  # 导入表格库
def map_platform_rfm(rfm_table):  # 定义平台已派生RFM表的映射接口
    if 'RFM' not in rfm_table.columns: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 缺列时停止
    mapped=rfm_table.copy()  # 保留平台表原始字段
    mapped['RFM']=mapped['RFM'].astype('string').str.strip()  # 对齐平台清洗步骤
    if mapped.empty or mapped['RFM'].isna().any(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 无有效标签时停止
    level_map={'高高高':'重要客户','高低高':'低购买频率客户','低高高':'长期未消费客户','低低高':'长期未消费低购买频率客户','高高低':'低消费客户','高低低':'低购买频率低消费客户','低高低':'长期未消费低消费客户'}  # 用字典表达同一组 if-elif 分类规则
    unknown_labels=mapped.loc[~mapped['RFM'].isin(level_map),'RFM'].value_counts(dropna=False)  # 单列报告未覆盖标签
    mapped['用户等级']=mapped['RFM'].map(level_map).fillna('其他待发展客户')  # 用字典得到平台同口径结果
    counts=mapped['用户等级'].value_counts()  # 生成柱图与饼图共同输入
    if counts.sum()!=len(mapped): raise AssertionError('RFM count reconciliation failed')  # 核对分类行数守恒
    return mapped,counts,unknown_labels  # 返回映射明细、计数与未知标签检查

from pathlib import Path  # 导入路径工具绑定规定资产
import pandas as pd  # 导入表格库以读取并标准化HDF字段
import numpy as np  # 导入近似数值核对工具
root=Path('/home/ubuntu/r2_data_mount/data/stock')  # 指定股票数据目录
basic_path,price_path=root/'stock_basic_data.h5',root/'stock_price_post_adjusted.parquet'  # 绑定基础信息与可按证券过滤的复权行情快照
if not basic_path.exists() or not price_path.exists(): raise FileNotFoundError('未找到课程股票数据文件，请从课程数据下载入口获取')  # 提示读者下载文件
basic=pd.read_hdf(basic_path,key='stock_basic_info',columns=['order_book_id','symbol','province']).rename(columns={'order_book_id':'raw_code','symbol':'name','province':'area'})  # 只读取公司筛选字段并标准化
basic['ts_code']=basic['raw_code'].str.replace('.XSHG','.SH',regex=False).str.replace('.XSHE','.SZ',regex=False)  # 统一证券代码后缀
basic['area']=basic['area'].astype('string').str.replace(r'[省市]$','',regex=True)  # 统一省市简称
required_basic={'ts_code','name','area'}  # 定义基础信息字段结构
required_price={'ts_code','trade_date','amount'}  # 定义行情覆盖字段结构
if not required_basic<=set(basic.columns): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 标准化后仍缺列时停止
focus=basic.loc[basic['area'].isin(['上海','江苏','浙江','安徽'])].sort_values('ts_code').head(40)  # 按固定规则选取长三角示例公司
if focus.empty: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 空对象时停止
focus_raw_codes=focus['raw_code'].tolist()  # 保留HDF原始代码以高效筛选行情
prices=pd.read_parquet(price_path,columns=['order_book_id','date','total_turnover'],filters=[('order_book_id','in',focus_raw_codes)]).rename(columns={'order_book_id':'ts_code','date':'trade_date','total_turnover':'amount'})  # 在读取层按对象过滤并标准化成交额
prices['ts_code']=prices['ts_code'].str.replace('.XSHG','.SH',regex=False).str.replace('.XSHE','.SZ',regex=False)  # 统一行情证券代码后缀
prices=prices.loc[pd.to_datetime(prices['trade_date']).between('2023-01-01','2025-12-31')].copy()  # 固定三年示例观察窗
if not required_price<=set(prices.columns): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 标准化后仍缺列时停止
market=prices.merge(focus,on='ts_code',how='inner',validate='many_to_one').copy()  # 连接行情与公司标签

market['trade_date']=pd.to_datetime(market['trade_date'])  # 统一交易日期
if focus.empty or market.empty: raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 空对象或空连接时停止
if market.duplicated(['ts_code','trade_date']).any(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 重复行情键时停止
window_end=market['trade_date'].max()  # 固定观察窗末日
coverage=market.groupby(['ts_code','name','area']).agg(recency=('trade_date',lambda dates:(window_end-dates.max()).days),frequency=('trade_date','nunique'),monetary=('amount','sum')).reset_index()  # 构造证券覆盖三指标
if len(coverage)<4 or not np.isfinite(coverage[['recency','frequency','monetary']].to_numpy()).all(): raise ValueError('输入文件、字段、样本量或数值不符合当前分析要求，请按本页说明检查')  # 四档分层与有限值检查要求
coverage['coverage_tier']=pd.qcut(coverage['frequency'].rank(method='first'),q=4,labels=['低','中低','中高','高'])  # 按活跃交易日分位分层
assert np.isclose(coverage['monetary'].sum(),market['amount'].sum(),rtol=1e-10,atol=0.01)  # 以货币容差核对成交额守恒
print(coverage.groupby(['area','coverage_tier'],observed=True).agg(securities=('ts_code','nunique'),turnover=('monetary','sum')))  # 输出地区分层规模与分母